package zhang;

import app.BaseSqlApp;
import common.Constant;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import util.SQLUtil;

public class DwdTestQuestion extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTestQuestion().init(3011,2,"DwdTestQuestion");
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //1 读取ods数据
        readOdsDb(tEnv,"DwdTestQuestion");
/*
{
  "database": "edu",
  "table": "test_exam_question",
  "type": "insert",
  "ts": 1652667846,
  "xid": 25687,
  "xoffset": 2,
  "data": {
    "id": 37272,
    "exam_id": 2497,
    "paper_id": 211,
    "question_id": 1798,
    "user_id": 307,
    "answer": "5874",
    "is_correct": "1",
    "score": 16.70,
    "create_time": "2022-02-22 22:08:41",
    "update_time": null,
    "deleted": "0"
  }
}
 */
        // 2 过滤测验问题表
        Table testExamQuestion = tEnv.sqlQuery(
                "select " +
                        "data['id'] id, " +
                        "data['exam_id'] exam_id, " +
                        "data['paper_id'] paper_id, " +
                        "data['question_id'] question_id, " +
                        "data['user_id'] user_id, " +
                        "data['is_correct'] is_correct, " +
                        "data['create_time'] create_time, " +
                        "data['update_time'] update_time, " +
                        "ts " +
                        "from ods_db " +
                        "where `database`='edu' " +
                        "and `table`='test_exam_question' " +
                        "and `type`='insert' ");
        tEnv.createTemporaryView("test_exam_question", testExamQuestion);


        // 3. 写出到kafka中
        tEnv.executeSql(
                "create table dwd_test_question( " +
                        "id string, " +
                        "exam_id string, " +
                        "paper_id string, " +
                        "question_id string, " +
                        "user_id string, " +
                        "is_correct string, " +
                        "create_time string, " +
                        "update_time string, " +
                        "ts bigint " +
                        ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TEST_QUESTION)
        );

        testExamQuestion.executeInsert("dwd_test_question");

    }
}
